package net.bwie.jtp.dws.log.funcion;

import akka.dispatch.Foreach;
import net.bwie.jtp.dws.log.bean.PageViewBean;
import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import scala.collection.Iterable;
import scala.collection.JavaConverters;

/**
 * @author: 🌹帅的一塌糊涂🌹
 * @description:
 * @params: 创建窗口函数,对每分钟窗口中数据进行计算 
 * @return:  
 * @date: 2025/5/19 18:24
 */
public class PageViewWindowFunction implements WindowFunction<PageViewBean, String, String, TimeWindow> {
    // TODO: 2025/5/19 格式化实例
    private FastDateFormat format = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss");

    @Override
    public void apply(String key, TimeWindow window, java.lang.Iterable<PageViewBean> input, Collector<String> out) throws Exception {

        // TODO: 2025/5/19 获取开时间和结束时间
        String stratTime = format.format(window.getStart());
        String endTime = format.format(window.getEnd());

        // TODO: 2025/5/19 对窗口中的数据进行计算
        long pvCount = 0L;
        long pvDuringTime = 0L;
        long uvCount = 0L;
        long sessionCount = 0L;
        // TODO: 2025/5/19 遍历每个PageViewBean对象，累加统计指标
        for (PageViewBean pageViewBean : input) {
            pvCount += pageViewBean.getPvCount();
            pvDuringTime += pageViewBean.getPvDuringTime();
            uvCount += pageViewBean.getUvCount();
            sessionCount += pageViewBean.getSessionCount();
        }
        // TODO: 2025/5/19 输出
        String output = stratTime+","+endTime+","+key+","+pvCount
                +","+uvCount+","+pvDuringTime+","+sessionCount+","+ System.currentTimeMillis();
        out.collect(output);
    }
}
